package com.im.business.api.offline;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import redis.clients.jedis.Jedis;

import java.util.Arrays;

/**
 * @Created leijun
 * @Date 2021/10/26 10:17 下午
 */
public class Application {

    private static Jedis jedis;

    public static void main(String[] args) throws InterruptedException {
        jedis = JedisManager.getJedis();
        KafkaManager kafkaManager = KafkaManager.getInstance();
        KafkaConsumer<String, String> consumer = kafkaManager.getConsumer();
        consumer.subscribe(Arrays.asList("single_offline_message", "group_offline_message"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(20);
            for (ConsumerRecord record : records) {
                String topic = record.topic();
                JSONObject message = JSONObject.parseObject(record.value().toString());
                if (topic.equals("single_offline_message")) {
                    handleSingleOfflineMessage(message);
                } else if (topic.equals("group_offline_message")) {
                    handleGroupOfflineMessage(message);
                }
            }
            Thread.sleep(1000);
        }
    }

    /**
     * 处理单聊离线消息
     * @param message
     */
    private static void handleSingleOfflineMessage(JSONObject message) {
        String receiverId = message.getString("receiverId");
        Long timestamp = message.getLong("timestamp");
        String key = "offline::single_message::" + receiverId;
        jedis.zadd(key, timestamp, message.toJSONString());
    }

    /**
     * 处理群离线消息
     * @param message
     */
    private static void handleGroupOfflineMessage(JSONObject message) {
        Long groupId = message.getLong("groupId");
        String receiverId = message.getString("receiverId");
        Long timestamp = message.getLong("timestamp");
        String key = "offline::group_message::" + groupId + ":" + receiverId;
        jedis.zadd(key, timestamp, message.toJSONString());
    }
}
